package com.zyx.flinkdemo.cdc.sql;

import com.zyx.flinkdemo.pojo.User;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/24 16:12
 * desc: 基于cdc的flink sql
 */
public class SqlCdcDemo {
    public static void main(String[] args) throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2.使用SQL方式读取MySQL变化数据
        tableEnv.executeSql("create table myuser(id int,name string,age int) with(" +
                "  'connector' = 'mysql-cdc', " +
                "  'hostname' = '10.1.207.12', " +
                "  'port' = '3306', " +
                "  'username' = 'root', " +
                "  'password' = 'root', " +
                "  'database-name' = 'db_report_amis', " +
                "  'table-name' = 'test' " +
                ")");

        // 3.转换为流打印
        Table table = tableEnv.sqlQuery("select * from myuser");
        tableEnv.toRetractStream(table, User.class).print();

        // 4.启动
        env.execute();
    }
}
